package com.open.source.kafka;

import com.open.source.MqSwitchControlConditionInitializer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication;
import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;

/**
 * @author ZonLen since on 2022/8/27 下午3:26
 */
@Slf4j
@Configuration
@ConditionalOnWebApplication
@AutoConfigureAfter(KafkaAutoConfiguration.class)
@ConditionalOnProperty(name = MqSwitchControlConditionInitializer.DEFAULT_MQ_SELECTED, havingValue = MqSwitchControlConditionInitializer.KAFKA)
public class KafkaConfiguration {

  @Bean
  public KafkaCompositeRecordInterceptorChain kafkaCompositeRecordInterceptorChain() {
    return new KafkaCompositeRecordInterceptorChain();
  }

  @Bean
  public TraceRecordInterceptor traceRecordInterceptor() {
    return new TraceRecordInterceptor();
  }

  @Bean
  @ConditionalOnBean({ConcurrentKafkaListenerContainerFactoryConfigurer.class,
      ConsumerFactory.class})
  public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
      ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
      ConsumerFactory<Object, Object> consumerFactory) {
    ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory);
    factory.setRecordInterceptor(kafkaCompositeRecordInterceptorChain());
    configurer.configure(factory, consumerFactory);
    return factory;
  }

}
